/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.iotdb.tsfile.read.filter;

import org.apache.iotdb.tsfile.read.common.TimeRange;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.apache.iotdb.tsfile.read.filter.factory.FilterType;
import org.apache.iotdb.tsfile.read.filter.operator.*;

import java.util.*;
import java.util.stream.Collectors;

public class TimeFilter {

    private TimeFilter() {
    }

    public static TimeEq eq(long value) {
        return new TimeEq(value);
    }

    public static TimeGt gt(long value) {
        return new TimeGt(value);
    }

    public static TimeGtEq gtEq(long value) {
        return new TimeGtEq(value);
    }

    public static TimeLt lt(long value) {
        return new TimeLt(value);
    }

    public static TimeLtEq ltEq(long value) {
        return new TimeLtEq(value);
    }

    public static TimeNotFilter not(Filter filter) {
        return new TimeNotFilter(filter);
    }

    public static TimeNotEq notEq(long value) {
        return new TimeNotEq(value);
    }

    public static TimeIn in(Set<Long> values, boolean not) {
        return new TimeIn(values, not);
    }

    public static TimeBetween between(long value1, long value2, boolean not) {
        return new TimeBetween(value1, value2, not);
    }

    public static class TimeBetween extends Between {

        private TimeBetween(long value1, long value2, boolean not) {
            super(value1, value2, FilterType.TIME_FILTER, not);
        }

        @Override
        public List<TimeRange> getTimeRanges() {
            long left = (long) value1;
            long right = (long) value2;
            if (not) {
                List<TimeRange> res = new ArrayList<>();
                if (left != Long.MIN_VALUE) {
                    res.add(new TimeRange(Long.MIN_VALUE, left - 1));
                }
                if (right != Long.MAX_VALUE) {
                    res.add(new TimeRange(right + 1, Long.MAX_VALUE));
                }
                return res;
            } else {
                return Collections.singletonList(new TimeRange(left, right));
            }
        }
    }

    public static class TimeIn extends In {

        private TimeIn(Set<Long> values, boolean not) {
            super(values, FilterType.TIME_FILTER, not);
        }

        @Override
        public List<TimeRange> getTimeRanges() {
            return ((Set<Long>) values)
                    .stream()
                    .map(
                            l -> {
                                long time = l;
                                return new TimeRange(time, time);
                            })
                    .collect(Collectors.toList());
        }
    }

    public static class TimeEq extends Eq {

        private TimeEq(long value) {
            super(value, FilterType.TIME_FILTER);
        }

        @Override
        public List<TimeRange> getTimeRanges() {
            return Collections.singletonList(new TimeRange((long) value, (long) value));
        }
    }

    public static class TimeNotEq extends NotEq {

        private TimeNotEq(long value) {
            super(value, FilterType.TIME_FILTER);
        }

        @Override
        public List<TimeRange> getTimeRanges() {
            long time = (long) value;
            if (time == Long.MIN_VALUE) {
                return Collections.singletonList(new TimeRange(time + 1, Long.MAX_VALUE));
            } else if (time == Long.MAX_VALUE) {
                return Collections.singletonList(new TimeRange(Long.MIN_VALUE, time - 1));
            } else {
                return Arrays.asList(
                        new TimeRange(Long.MIN_VALUE, time - 1), new TimeRange(time + 1, Long.MAX_VALUE));
            }
        }
    }

    public static class TimeGt extends Gt {

        private TimeGt(long value) {
            super(value, FilterType.TIME_FILTER);
        }

        @Override
        public List<TimeRange> getTimeRanges() {
            long left = (long) value;
            if (left != Long.MAX_VALUE) {
                return Collections.singletonList(new TimeRange(left + 1, Long.MAX_VALUE));
            } else {
                return Collections.emptyList();
            }
        }
    }

    public static class TimeGtEq extends GtEq {

        private TimeGtEq(long value) {
            super(value, FilterType.TIME_FILTER);
        }

        @Override
        public List<TimeRange> getTimeRanges() {
            return Collections.singletonList(new TimeRange((long) value, Long.MAX_VALUE));
        }
    }

    public static class TimeLt extends Lt {

        private TimeLt(long value) {
            super(value, FilterType.TIME_FILTER);
        }

        @Override
        public List<TimeRange> getTimeRanges() {
            long right = (long) value;
            if (right != Long.MIN_VALUE) {
                return Collections.singletonList(new TimeRange(Long.MIN_VALUE, right - 1));
            } else {
                return Collections.emptyList();
            }
        }
    }

    public static class TimeLtEq extends LtEq {

        private TimeLtEq(long value) {
            super(value, FilterType.TIME_FILTER);
        }

        @Override
        public List<TimeRange> getTimeRanges() {
            return Collections.singletonList(new TimeRange(Long.MIN_VALUE, (long) value));
        }
    }

    public static class TimeNotFilter extends NotFilter {

        private TimeNotFilter(Filter filter) {
            super(filter);
        }
    }

    /**
     * returns a default time filter by whether it's an ascending query.
     *
     * <p>If the data is read in descending order, we use the largest timestamp to set to the filter,
     * so the filter should be TimeLtEq. If the data is read in ascending order, we use the smallest
     * timestamp to set to the filter, so the filter should be TimeGtEq.
     */
    public static Filter defaultTimeFilter(boolean ascending) {
        return ascending ? TimeFilter.gtEq(Long.MIN_VALUE) : TimeFilter.ltEq(Long.MAX_VALUE);
    }
}
